﻿using Microsoft.Extensions.Hosting;
using MySqlSync.Core.Interfaces;
using MySqlSync.Core.Models;
using System.Text.Json;

namespace MySqlSync.SyncService
{
    /// <summary>
    /// 同步消费者服务
    /// </summary>
    public class SyncConsumerService : ISyncConsumerService, IHostedService
    {
        private readonly IKafkaConsumerService _kafkaConsumer;
        private readonly IDbService _dbService;
        private readonly string _targetConnectionString;
        private int _retryCount = 3;
        private int _retryDelayMs = 1000;

        /// <summary>
        /// 构造函数
        /// </summary>
        public SyncConsumerService(IKafkaConsumerService kafkaConsumer, IDbService dbService, SyncConfig config)
        {
            _kafkaConsumer = kafkaConsumer ?? throw new ArgumentNullException(nameof(kafkaConsumer));
            _dbService = dbService ?? throw new ArgumentNullException(nameof(dbService));

            if (config == null)
                throw new ArgumentNullException(nameof(config));

            if (string.IsNullOrEmpty(config.TargetConnectionString))
                throw new ArgumentException("目标数据库连接字符串不能为空", nameof(config));

            _targetConnectionString = config.TargetConnectionString;
        }

        /// <summary>
        /// 启动消费者服务
        /// </summary>
        public async Task StartAsync(CancellationToken cancellationToken)
        {
            Console.WriteLine("同步消费者服务已启动");

            // 开始消费消息
            await _kafkaConsumer.ConsumeAsync(HandleSyncRecordAsync, cancellationToken);

            Console.WriteLine("同步消费者服务已停止");
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            // 添加资源释放逻辑
            return Task.Run(() => StopConsumer(), cancellationToken);
        }

        /// <summary>
        /// 处理同步记录
        /// </summary>
        private async Task HandleSyncRecordAsync(SyncRecord record)
        {
            // 使用重试机制处理记录
            await RetryAsync(async () =>
            {
                try
                {
                    //Console.WriteLine($"处理记录输出完整记录: {JsonSerializer.Serialize(record)}"); // 输出完整记录

                    Console.WriteLine($"处理记录: {record.TableName} - {record.RecordId} - {record.Operation}");
                    
                    // 应用同步记录到目标数据库
                    await _dbService.ApplySyncRecordAsync(record, _targetConnectionString);

                    Console.WriteLine($"记录处理成功: {record.TableName} - {record.RecordId}");
                    return true;
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"处理记录失败 {record.TableName} - {record.RecordId}: {ex.Message}");
                    throw;
                }
            });
        }

        /// <summary>
        /// 带重试的异步操作
        /// </summary>
        private async Task RetryAsync(Func<Task<bool>> operation)
        {
            int attempts = 0;

            while (attempts < _retryCount)
            {
                attempts++;
                try
                {
                    if (await operation())
                    {
                        return;
                    }
                }
                catch
                {
                    if (attempts >= _retryCount)
                    {
                        throw;
                    }

                    Console.WriteLine($"将在 {_retryDelayMs}ms 后重试 (第 {attempts}/{_retryCount} 次)");
                    await Task.Delay(_retryDelayMs);
                }
            }
        }

        /// <summary>
        /// 停止消费者并释放相关资源
        /// </summary>
        private void StopConsumer()
        {
            try
            {
                Console.WriteLine("正在停止Kafka消费者...");

                // 调用Kafka消费者的停止方法
                if (_kafkaConsumer != null)
                {
                    //_kafkaConsumer.StopAsync(); // 同步停止消费者
                    Console.WriteLine("Kafka消费者已停止");
                }

                Console.WriteLine("消费者资源释放完成");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"停止消费者过程中发生错误: {ex.Message}");
                // 可根据需要决定是否重新抛出异常
                // throw;
            }
        }
    }
}
